package org.qiyu.live.id.generate.provider.service.impl;

import jakarta.annotation.Resource;
import org.qiyu.live.id.generate.provider.dao.mapper.IdGenerateMapper;
import org.qiyu.live.id.generate.provider.dao.po.IdBuilderPO;
import org.qiyu.live.id.generate.provider.service.bo.LocalSeqIdBO;
import org.qiyu.live.id.generate.provider.service.bo.LocalUnSeqIdBO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.qiyu.live.id.generate.provider.service.IdGenerateService;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

@Service
public class IdGenerateServiceImpl implements IdGenerateService, InitializingBean {

    private final static float UPDATE_RATE = 0.75f;

    private final static float UN_UPDATE_RATE = 0.25f;

    public static final int IS_SEQ = 0;  //0 有序

    private final static Logger LOGGER = LoggerFactory.getLogger(IdGenerateServiceImpl.class);

    private static final Map<Integer, LocalSeqIdBO> localSeqIdBOMap = new ConcurrentHashMap<>();

    private static final Map<Integer, LocalUnSeqIdBO> localUnSeqIdBOMap = new ConcurrentHashMap<>();

    private static final Map<Integer,Semaphore> semaphoreMap = new ConcurrentHashMap<>();

    //创建线程池
    private static final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8,16,3, TimeUnit.SECONDS,new ArrayBlockingQueue<>(1000),new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("id-generate-provider");
            return thread;
        }
    });

    @Resource
    IdGenerateMapper idGenerateMapper;

    //有序id
    @Override
    public Long getSeqId(Integer id) {
        if (id == null) {
            LOGGER.error("[getSeqId] id is null,id is {}",id);
            return null;
        }

        //从本地缓存中取数，若是初始化没有加载出来，则返回空
        LocalSeqIdBO localSeqIdBO = localSeqIdBOMap.get(id);
        if (localSeqIdBO == null) {
            LOGGER.error("[getSeqId] localSeqIdBO is null,id is {}",id);
            return null;
        }

        //当号码段使用超过75%时，触发异步刷新，提前更新号码段
        this.refreshLocalSeqId(localSeqIdBO);

        //产生一个自增的id
        long andIncrement = localSeqIdBO.getCurrentNum().getAndIncrement();

        //如果当前的id大于上限，则需要做一个保护机制，阻止继续自增占用其他的id段
        if(andIncrement > localSeqIdBO.getNextThreshold()){
            //若是获取速度过快，还来不及从数据库获取下一个号段，则等待少量时间，等数据库查询成功
            int maxMs = 15;
            for(int i = 1; i <= maxMs; i ++){
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                if(localSeqIdBOMap.get(id).getCurrentNum().get() <= localSeqIdBOMap.get(id).getNextThreshold()) {
                    break;
                }
                //15毫秒还未从数据库获取到则返回失败
                if(i == maxMs){
                    LOGGER.error("localSeqIdBO is null,id is {}",id);
                    return null;
                }
            }
        }

        return andIncrement;
    }

    //无序id
    @Override
    public Long getUnSeqId(Integer id) {
        if (id == null) {
            LOGGER.error("[getUnSeqId] id is null,id is {}",id);
            return null;
        }

        LocalUnSeqIdBO localUnSeqIdBO = localUnSeqIdBOMap.get(id);

        if (localUnSeqIdBO == null) {
            LOGGER.error("[getUnSeqId] localUnSeqIdBO is null,id is {}", id);
            return null;
        }

        Long andIncrement = localUnSeqIdBO.getIdQueue().poll();
        if(andIncrement == null){
            int maxMs = 30;
            for(int i = 1; i <= maxMs; i ++){
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                if(localUnSeqIdBO.getIdQueue().poll() != null) {
                    break;
                }
                //15毫秒还未从数据库获取到则返回失败
                if(i == maxMs){
                    LOGGER.error("localSeqIdBO is null,id is {}",id);
                    return null;
                }
            }
        }

        this.refreshLocalUnSeqId(localUnSeqIdBO);
        return andIncrement;
    }

    private void refreshLocalUnSeqId(LocalUnSeqIdBO localUnSeqIdBO) {
        Long begin = localUnSeqIdBO.getCurrentStart();
        Long end = localUnSeqIdBO.getNextThreshold();
        long size = localUnSeqIdBO.getIdQueue().size();  //队列中剩余的id数量
        if( size < (end - begin) * UN_UPDATE_RATE ){
            //用id去取Semaphore
            Semaphore semaphore = semaphoreMap.get(localUnSeqIdBO.getId());
            if (semaphore == null) {
                LOGGER.error("semaphore is null,id is {}",localUnSeqIdBO.getId());
                return;
            }
            //取到Semaphore后尝试进入该线程，如果返回true，则证明线程池中没有该线程正在运行
            boolean b = semaphore.tryAcquire();
            if (b) {
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try{
                            LOGGER.info("[refreshLocalUnSeqId] threadPoolExecutor is running");
                            IdBuilderPO idBuilderPO = idGenerateMapper.selectById(localUnSeqIdBO.getId());
                            noticeSQLRecord(idBuilderPO);
                        }catch (Exception e){
                            LOGGER.error("[refreshLocalUnSeqId] idGenerateMapper is error",e);
                        }finally {
                            semaphoreMap.get(localUnSeqIdBO.getId()).release();
                        }

                    }
                });
            }


        }
    }


    /**
     * 当号码段使用超过75%时，触发异步刷新，提前更新号码段
     */
    public void refreshLocalSeqId( LocalSeqIdBO localSeqIdBO) {
        long step = localSeqIdBO.getNextThreshold() - localSeqIdBO.getCurrentStart();
        if(localSeqIdBO.getCurrentNum().get() - localSeqIdBO.getCurrentStart() > UPDATE_RATE * step) {

            //用id去取Semaphore
            Semaphore semaphore = semaphoreMap.get(localSeqIdBO.getId());
            if (semaphore == null) {
                LOGGER.error("semaphore is null,id is {}",localSeqIdBO.getId());
                return;
            }
            //取到Semaphore后尝试进入该线程，如果返回true，则证明线程池中没有该线程正在运行
            boolean b = semaphore.tryAcquire();
            if (b) {
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try{
                            LOGGER.info("[refreshLocalSeqId] threadPoolExecutor is running");
                            IdBuilderPO idBuilderPO = idGenerateMapper.selectById(localSeqIdBO.getId());
                            noticeSQLRecord(idBuilderPO);
                        }catch (Exception e){
                            LOGGER.error("[refreshLocalSeqId] idGenerateMapper is error",e);
                        }finally {
                            semaphoreMap.get(localSeqIdBO.getId()).release();

                        }
                    }
                });
            }

        }

    }


    @Override
    public void afterPropertiesSet() throws Exception {
        List<IdBuilderPO> idBuilderPOS = idGenerateMapper.selectAll();
        for (IdBuilderPO idBuilderPO : idBuilderPOS) {
            noticeSQLRecord(idBuilderPO);
            //创建一个Semaphore放入map，并为其定义一个id，后续被标识这个id的线程，在线程池中只允许执行一次
            semaphoreMap.put(idBuilderPO.getId(), new Semaphore(1));
        }
    }

    //尝试标记某个号码段已经被占用
    public void noticeSQLRecord(IdBuilderPO idBuilderPO){
        //更新标识占用号码段，若失败（即其他服务同时更新时，导致version标识不一致）则重新尝试
        int i = idGenerateMapper.updateNewIdCountAndVersion(idBuilderPO.getId(), idBuilderPO.getVersion());
        if(i > 0){
            localHander(idBuilderPO);
        }
        if (i == 0) {
            //更新失败，重新查询最新的version，并尝试重新更新
            for(int m = 0; m < 3; m++){
                IdBuilderPO idBuilderPO1 = idGenerateMapper.selectById(idBuilderPO.getId());
                int i1 = idGenerateMapper.updateNewIdCountAndVersion(idBuilderPO1.getId(), idBuilderPO1.getVersion());
                if (i1 > 0) {
                    localHander(idBuilderPO);
                    return;
                }

            }
            throw new RuntimeException("id段占用失败, 竞争激烈, id :"+idBuilderPO.getId());
        }

    }

    public void localHander(IdBuilderPO idBuilderPO){
        long currentStart = idBuilderPO.getCurrentStart();
        long nextThreshold = idBuilderPO.getNextThreshold();
        long currentNum = currentStart;
        if(idBuilderPO.getIsSeq() == IS_SEQ){
            LocalSeqIdBO localSeqIdBO = new LocalSeqIdBO();
            localSeqIdBO.setCurrentNum(new AtomicLong(currentNum));
            localSeqIdBO.setId( idBuilderPO.getId() );
            localSeqIdBO.setCurrentStart(currentStart);
            localSeqIdBO.setNextThreshold(nextThreshold);
            localSeqIdBOMap.put(idBuilderPO.getId(), localSeqIdBO);
        }else{
            LocalUnSeqIdBO localUnSeqIdBO = new LocalUnSeqIdBO();
            localUnSeqIdBO.setId( idBuilderPO.getId() );
            localUnSeqIdBO.setCurrentStart(currentStart);
            localUnSeqIdBO.setNextThreshold(nextThreshold);
            long begin = localUnSeqIdBO.getCurrentStart();
            long end = localUnSeqIdBO.getNextThreshold();
            //将本地的id段打乱，放入到map中
            List<Long> list = new ArrayList<>();
            for (long j = begin; j <= end; j++){
                list.add(j);
            }
            ConcurrentLinkedQueue<Long> idQueue = new ConcurrentLinkedQueue<>();
            Collections.shuffle(list);
            idQueue.addAll(list);
            localUnSeqIdBO.setIdQueue(idQueue);
            localUnSeqIdBOMap.put(idBuilderPO.getId(), localUnSeqIdBO);
        }

    }
}
